/* Copyright  2009-2010
   @Author
   Richard Changde Yin            e-mail yinchangde@hotmail.com

   This program is free software; you can redistribute it and/or modify
   it under the terms of the GNU General Public License as published by
   the Free Software Foundation; version 2 of the License.

   This program is distributed in the hope that it will be useful,
   but WITHOUT ANY WARRANTY; without even the implied warranty of
   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
   GNU General Public License for more details.

   You should have received a copy of the GNU General Public License
   along with this program; if not, write to the Free Software
   Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA */

/**
 * @filename: mm_handler.cc
 */

#include <server_includes.h>

#include <mm_handler.h>

#include <error.h>
#include <table.h>
#include <session.h>
#include <current_session.h>
#include <field/timestamp.h>
#include <field/varstring.h>

#include <catalogtables.h>
#include <databaseManagerImpl.h>

#include <mutex.h>

using namespace BackEnd ;

handler *get_new_handler(TABLE_SHARE *table,MEM_ROOT *mem_root)

{
  return new (mem_root) mm_handler(table);
}
#define GET_DBMGR      Session* session = ha_session();                        \
                       DatabaseManager* dbmgr = session->getDatabaseManager();

/*****************************************************************************
** MM tables
*****************************************************************************/

mm_handler::mm_handler(TABLE_SHARE *table_arg):handler( table_arg),
                                               records_changed (0),
                                               key_stat_version(0),
                                               internal_table  (0),
                                               istemp_table    (0),
                                               start_of_scan(0)
{
  file = (MM_INFO*)malloc(sizeof(MM_INFO));
  bzero(file,sizeof(MM_INFO));
  cached_table_flags = this->table_flags();
}

mm_handler::~mm_handler()
{
  if(file)
  {
    if(file->lastkey)free(file->lastkey);
    file->lastkey = 0;
    if(file->table_alias) free(file->table_alias);
    file->table_alias =0;
    free((unsigned char*)file);
    file = 0;
  }
}
static const char *mm_handler_exts[] = {
  NULL
};

const char ** mm_handler::bas_ext() const
{
  return mm_handler_exts;
}

/*
  Hash index statistics is updated (copied from HP_KEYDEF::hash_buckets to
  rec_per_key) after 1/HEAP_STATS_UPDATE_THRESHOLD fraction of table records
  have been inserted/updated/deleted. delete_all_rows() and table flush cause
  immediate update.

  NOTE
   hash index statistics must be updated when number of table records changes
   from 0 to non-zero value and vice versa. Otherwise records_in_range may
   erroneously return 0 and 'range' may miss records.
*/
#define HEAP_STATS_UPDATE_THRESHOLD 10

int mm_handler::open(const char *, int mode, uint32_t test_if_locked)
{
   int ret   = 1 ;
   SimpleMutex _mutex(THR_LOCK_MM);
   if ( !table  ) return (1) ;

   GET_DBMGR   ;
   if ( test_if_locked & HA_OPEN_INTERNAL_TABLE ) internal_table = true ;

   if ( mode & HA_OPEN_TMP_TABLE ) return (1) ;


   assert(dbmgr);
   char* thefullname = table->full_table_name;
   if(!composeFullTableName(table->getOwner(),table->alias,
                            thefullname,IDENTIFIER_LENGTH))
     return (HA_ER_TOO_LONG_IDENT);

   if (!file){
     file = (MM_INFO*)malloc(sizeof(MM_INFO));
     if(!file) return (HA_ERR_OUT_OF_MEM);
     bzero(file,sizeof(MM_INFO));
   }

   if (!table->getChunkPtr() )
  {
     // Then we start to open table .
     ret = dbmgr->openTable(thefullname,table,false);
     if ( ret )
      // something wrong. logging
      return ret ;

  }else {
      /* get the backend open count */
      if ( 0 >= table->getBEOpenCount()) {
         /// open it without creating index.
         ret = dbmgr->openTable(thefullname,table, false) ;
         if(ret) return ret ;
      }
  }
   /* define ref_length*/
  ref_length= sizeof(unsigned char*);

  CTABLE* ctable = static_cast<CTABLE*>(table->getCTABLEPtr());
  file->current_record= UINT32_MAX;  /* No current record */
  file->lastinx = file->errkey= -1 ;
  file->lastkey = (unsigned char*)malloc(MAX_KEY_LENGTH);
  file->mode    = mode ;
  file->table_alias = strdup(thefullname);
  thr_lock_data_init(&ctable->lock,&file->lock,NULL);
  {
    /* Initialize variables for the opened table */
    set_keys_for_scanning();
    /*
      We cannot run update_key_stats() here because we do not have a
      lock on the table. The 'records' count might just be changed
      temporarily at this moment and we might get wrong statistics (Bug
      #10178). Instead we request for update. This will be done in
      mm_handler::info(), which is always called before key statistics are
      used.
    */
    //key_stat_version= file->s->key_stat_version-1;
  }
  return (0);
}

int mm_handler::close(void)
{
  SimpleMutex _mutex(THR_LOCK_MM);
  Session*  session      = ha_session();
  if ( NULL != session ) {
      DatabaseManager* dbmgr = session->getDatabaseManager() ;
      if(NULL != dbmgr)dbmgr->closeTable(table) ;
  }
  if(file)
  {
    if(file->table_alias)
    {
      free(file->table_alias);file->table_alias=0;
    }
    free ((unsigned char*) file);
  }
  file = 0 ;
  return (0);
}


/*
  Create a copy of this table

  DESCRIPTION
    Do same as default implementation but use file->s->name instead of
    table->s->path. This is needed by Windows where the clone() call sees
    '/'-delimited path in table->s->path, while ha_peap::open() was called
    with '\'-delimited path.
*/

handler *mm_handler::clone(MEM_ROOT *mem_root)
{
  handler *new_handler= get_new_handler(table->s, mem_root);

  if (new_handler && !new_handler->ha_open(table, table->alias, table->db_stat,
                                           HA_OPEN_IGNORE_IF_LOCKED))
    return new_handler;
  return NULL;  /* purecov: inspected */
}


const char * mm_handler::index_type(uint32_t)
{
  return "HASH" ;
}


uint32_t mm_handler::index_flags(uint32_t, uint32_t, bool) const
{
  /*
  return ((table_share->key_info[inx].algorithm == HA_KEY_ALG_BTREE) ?
          HA_READ_NEXT | HA_READ_PREV | HA_READ_ORDER | HA_READ_RANGE :
          HA_ONLY_WHOLE_INDEX | HA_KEY_SCAN_NOT_ROR);
   */

  return  (HA_READ_NEXT | HA_ONLY_WHOLE_INDEX | HA_KEY_SCAN_NOT_ROR);
}


/*
  Compute which keys to use for scanning

  SYNOPSIS
    set_keys_for_scanning()
    no parameter

  DESCRIPTION
    Set the bitmap btree_keys, which is used when the upper layers ask
    which keys to use for scanning. For each btree index the
    corresponding bit is set.

  RETURN
    void
*/

void mm_handler::set_keys_for_scanning(void)
{
  btree_keys.clear_all();
  for (uint32_t i= 0 ; i < table->s->keys ; i++)
  {
    //if (table->key_info[i].algorithm == HA_KEY_ALG_BTREE)
       btree_keys.set_bit(i);
  }
}


void mm_handler::update_key_stats()
{
  for (uint32_t i= 0; i < table->s->keys; i++)
  {
    KEY *key=table->key_info+i;
    if (!key->rec_per_key)
      continue;
    if (key->algorithm != HA_KEY_ALG_BTREE)
    {
      if (key->flags & HA_NOSAME)
        key->rec_per_key[key->key_parts-1]= 1;
      else
      {/*
        ha_rows hash_buckets= file->s->keydef[i].hash_buckets;
        uint32_t no_records= hash_buckets ? (uint) (file->s->records/hash_buckets) : 2;
        if (no_records < 2)
          no_records= 2;
        key->rec_per_key[key->key_parts-1]= no_records;
       */
      }
    }
  }
  records_changed= 0;
  /* At the end of update_key_stats() we can proudly claim they are OK. */
  //key_stat_version= file->s->key_stat_version;
}


int mm_handler::write_row(unsigned char * buf)
{
  int res=0;
  ha_statistic_increment(&SSV::ha_write_count);
  GET_DBMGR   ;
  if (table->next_number_field && buf == table->record[0])
  {
    if ((res= update_auto_increment()))
      return res;
  }
  if  (!dbmgr) res = ER_INVALID_SESSION_STATE ;
  else {
    DatabaseManagerImpl* dbmgrimpl = (DatabaseManagerImpl*) dbmgr ;
    /// insert new record.
    res =  dbmgrimpl->newRecord(*table,buf) ;
  }
  /*
  if (!res && (++records_changed*HEAP_STATS_UPDATE_THRESHOLD >
               file->s->records))
  {

       We can perform this safely since only one writer at the time is
       allowed on the table.

    file->s->key_stat_version++;

    key_stat_version ++ ;
  }*/
  return res;
}

int mm_handler::update_row(const unsigned char * , unsigned char * )
{
  int res = 0;
  ha_statistic_increment(&SSV::ha_update_count);
  if (table->timestamp_field_type & TIMESTAMP_AUTO_SET_ON_UPDATE)
    table->timestamp_field->set_time();
  //res= heap_update(file,old_data,new_data);
  //if (!res && ++records_changed*HEAP_STATS_UPDATE_THRESHOLD >
  //            file->s->records)
  {
    /*
       We can perform this safely since only one writer at the time is
       allowed on the table.
    */
    //file->s->key_stat_version++;
    key_stat_version ++ ;
  }
  return res;
}

int mm_handler::delete_row(const unsigned char * record)
{
  int res=0;
  ha_statistic_increment(&SSV::ha_delete_count);
  GET_DBMGR   ;
  if(!dbmgr)  return (1);
  res = (int)dbmgr->delete_row(*table,file,( unsigned char *)record);
  {
    /*
       We can perform this safely since only one writer at the time is
       allowed on the table.
    */
    //file->s->key_stat_version++;
    key_stat_version ++ ;
  }
  return res;
}

int mm_handler::index_read_map(unsigned char* buf, const unsigned char * key,
                               key_part_map keypart_map,
                               enum ha_rkey_function find_flag)
{
  int error =0;
  assert(inited==INDEX);
  GET_DBMGR   ;
  ha_statistic_increment(&SSV::ha_read_key_count);
  error = dbmgr->mm_rkey(*table,file,buf,active_index, key, keypart_map, find_flag);
  table->status = error ? STATUS_NOT_FOUND : 0;
  return error;
}

int mm_handler::index_read_last_map(unsigned char * buff, const
                                    unsigned char * key,
                                    key_part_map keypart_map)
{
  assert(inited==INDEX);
  ha_statistic_increment(&SSV::ha_read_key_count);
  GET_DBMGR   ;
  int error= dbmgr->mm_rkey(*table,file, buff, active_index, key, keypart_map,
                            HA_READ_PREFIX_LAST);
  table->status= error ? STATUS_NOT_FOUND : 0;
  return error;
}

int mm_handler::index_read_idx_map(unsigned char* buff, uint32_t index,
                                const unsigned char * key,
                                key_part_map keypart_map,
                                enum ha_rkey_function find_flag)
{
  ha_statistic_increment(&SSV::ha_read_key_count);
  Session* session = ha_session();
  DatabaseManager* dbmgr = session->getDatabaseManager();
  int error =dbmgr->mm_rkey(*table,file, buff, index, key, keypart_map,
                            find_flag);
  table->status = error ? STATUS_NOT_FOUND : 0;
  return error;
}

int mm_handler::index_next(unsigned char * )
{
  assert(inited==INDEX);
  ha_statistic_increment(&SSV::ha_read_next_count);
  int error=0;//heap_rnext(file,buf);
  table->status=error ? STATUS_NOT_FOUND: 0;
  return error;
}

int mm_handler::index_prev(unsigned char * )
{
  assert(inited==INDEX);
  ha_statistic_increment(&SSV::ha_read_prev_count);
  int error=0;//heap_rprev(file,buf);
  table->status=error ? STATUS_NOT_FOUND: 0;
  return error;
}

int mm_handler::index_first(unsigned char * buf)
{
  assert(inited==INDEX);
  ha_statistic_increment(&SSV::ha_read_first_count);
  Session* session = ha_session();
  DatabaseManager* dbmgr = session->getDatabaseManager();
  int error=dbmgr->mm_rfirst(*table,file, buf, active_index);
  table->status=error ? STATUS_NOT_FOUND: 0;
  return error;
}

int mm_handler::index_last(unsigned char * )
{
  assert(inited==INDEX);
  ha_statistic_increment(&SSV::ha_read_last_count);
  int error=0;//=heap_rlast(file, buf, active_index);
  table->status=error ? STATUS_NOT_FOUND: 0;
  return error;
}

int mm_handler::index_next_same(unsigned char *,
                               const unsigned char *,
                               uint32_t )
{
  int error;
  assert(inited==INDEX);
  ha_statistic_increment(&SSV::ha_read_next_count);
  do
  {
    //error= mi_rnext_same(file,buf);
    error =0;
  } while (error == HA_ERR_RECORD_DELETED);
  table->status=error ? STATUS_NOT_FOUND: 0;
  return error;
}

int mm_handler::rnd_init(bool scan)
{
  return scan ? mm_scan_init(file) : 0;
}

int mm_handler::mm_scan_init(MM_INFO *_info)
{
  _info->lastinx= -1;
  _info->current_record= UINT32_MAX; /* No current record */
  _info->update=0;
  _info->next_block=0;
  start_of_scan = 1;
  return(0);
}
int mm_handler::rnd_next(unsigned char * )
{
  ha_statistic_increment(&SSV::ha_read_rnd_next_count);
  int error = 0;
  //int error=heap_scan(file, buf);
  table->status=error ? STATUS_NOT_FOUND: 0;
  return error;
}

int mm_handler::rnd_pos(unsigned char* buff, unsigned char* pos)
{
  int error =0;
  GET_DBMGR;
  ha_statistic_increment(&SSV::ha_read_rnd_count);

  file->lastinx  = -1 ;

  if (!(file->current_ptr= pos))
  {
    file->update= 0;
    return(my_errno= HA_ERR_END_OF_FILE);
  }
  /*Todo: get pageInfo.*/


  file->update=HA_STATE_PREV_FOUND | HA_STATE_NEXT_FOUND | HA_STATE_AKTIV;
  ((DatabaseManagerImpl*)dbmgr)->mm_extract_record((CTABLE*)table->getCTABLEPtr(),
                                                   buff,
                                         (const unsigned char*)file->current_ptr);
  table->status=error ? STATUS_NOT_FOUND: 0;
  return error;
}

void mm_handler::position(const unsigned char *)
{
  unsigned char* curr_pos;
  curr_pos  = ((file->update & HA_STATE_AKTIV) ? file->current_ptr :
                                                (unsigned char*) 0);
  ref = curr_pos;
}

int mm_handler::info(uint32_t flag)
{
  int res = 0;
  GET_DBMGR;
  if  (!dbmgr || !file) res = ER_INVALID_SESSION_STATE ;

  dbmgr->getInfo(file);

  if (flag & HA_STATUS_AUTO)
  {
    stats.auto_increment_value= file->auto_increment;
  }

  if (flag & HA_STATUS_ERRKEY)
  {
    errkey = (uint32_t )((DatabaseManagerImpl*)dbmgr)->errkey ;
  }
  stats.records=              file->hlderinfo.records;
  stats.deleted=              file->hlderinfo.deleted;
  stats.mean_rec_length=      file->hlderinfo.reclength;
  stats.data_file_length=     file->hlderinfo.data_length;
  stats.index_file_length=    file->hlderinfo.index_length;
  stats.max_data_file_length= (uint64_t)~0;

  /*

    If info() is called for the first time after open(), we will still
    have to update the key statistics. Hoping that a table lock is now
    in place.

  if (key_stat_version != file->s->key_stat_version)
    update_key_stats();
  */
  return 0;
}


enum row_type mm_handler::get_row_type() const
{
  /*
  if (file->s->recordspace.is_variable_size)
    return ROW_TYPE_DYNAMIC;
  */
  return ROW_TYPE_FIXED;
}

int mm_handler::extra(enum ha_extra_function )
{
 // return heap_extra(file,operation);
    return 0;
}


int mm_handler::reset()
{
  file->lastinx          = -1;
  file->errkey           = -1;
  file->current_record   = UINT32_MAX;
  file->current_ptr      = 0;
  file->current_hIdxNode = 0;
  //file->current_hash_ptr=0;
  file->update=0;
  file->next_block=0;
  return 0;
}


int mm_handler::delete_all_rows()
{
  if (table->s->tmp_table == NO_TMP_TABLE)
  {
    /*
       We can perform this safely since only one writer at the time is
       allowed on the table.
    */
   // file->>key_stat_version++;
  }
  Session*  session = ha_session();
  DatabaseManager* dbmgr = session->getDatabaseManager() ;
  return (int)dbmgr->delete_all_rows(*table);
}

int mm_handler::external_lock(Session *, int)
{
  return 0;               // No external locking
}


/*
  Disable indexes.

  SYNOPSIS
    disable_indexes()
    mode        mode of operation:
                HA_KEY_SWITCH_NONUNIQ      disable all non-unique keys
                HA_KEY_SWITCH_ALL          disable all keys
                HA_KEY_SWITCH_NONUNIQ_SAVE dis. non-uni. and make persistent
                HA_KEY_SWITCH_ALL_SAVE     dis. all keys and make persistent

  DESCRIPTION
    Disable indexes and clear keys to use for scanning.

  IMPLEMENTATION
    HA_KEY_SWITCH_NONUNIQ       is not implemented.
    HA_KEY_SWITCH_NONUNIQ_SAVE  is not implemented with HEAP.
    HA_KEY_SWITCH_ALL_SAVE      is not implemented with HEAP.

  RETURN
    0  ok
    HA_ERR_WRONG_COMMAND  mode not implemented.
*/

int mm_handler::disable_indexes(uint32_t mode)
{
  int error;
  Session*  session = ha_session();
  DatabaseManager* dbmgr = session->getDatabaseManager() ;

  if (mode == HA_KEY_SWITCH_ALL)
  {
    if ((error=dbmgr->disableIndex(*table,table->getChunkPtr())))
    {
      return (error);
    }
    error = 0;
  }
  else
  {
    /* mode not implemented */
    error= HA_ERR_WRONG_COMMAND;
  }
  return error;
}


/*
  Enable indexes.

  SYNOPSIS
    enable_indexes()
    mode        mode of operation:
                HA_KEY_SWITCH_NONUNIQ      enable all non-unique keys
                HA_KEY_SWITCH_ALL          enable all keys
                HA_KEY_SWITCH_NONUNIQ_SAVE en. non-uni. and make persistent
                HA_KEY_SWITCH_ALL_SAVE     en. all keys and make persistent

  DESCRIPTION
    Enable indexes and set keys to use for scanning.
    The indexes might have been disabled by disable_index() before.
    To be sure, call handler::delete_all_rows() before.

  IMPLEMENTATION
    HA_KEY_SWITCH_NONUNIQ       is not implemented.
    HA_KEY_SWITCH_NONUNIQ_SAVE  is not implemented with HEAP.
    HA_KEY_SWITCH_ALL_SAVE      is not implemented with HEAP.

  RETURN
    0  ok
    HA_ERR_CRASHED  data or index is non-empty. Delete all rows and retry.
    HA_ERR_WRONG_COMMAND  mode not implemented.
*/

int mm_handler::enable_indexes(uint32_t mode)
{
  int error;
  Session*  session = ha_session();
  DatabaseManager* dbmgr = session->getDatabaseManager() ;
  if (mode == HA_KEY_SWITCH_ALL)
  {
     if ((error= dbmgr->disableIndex(*table,table->getChunkPtr())))
     {
       return (error);
     }
      error = 0;
  }
  else
  {
    /* mode not implemented */
    error= HA_ERR_WRONG_COMMAND;
  }
  return error;
}


/*
  Test if indexes are disabled.

  SYNOPSIS
    indexes_are_disabled()
    no parameters

  RETURN
    0  indexes are not disabled
    1  all indexes are disabled
   [2  non-unique indexes are disabled - NOT YET IMPLEMENTED]
*/

int mm_handler::indexes_are_disabled(void)
{
    Session*  session      = ha_session();
    DatabaseManager* dbmgr = session->getDatabaseManager() ;

    if (OK == dbmgr->is_indexes_are_disabled(*table,table->getChunkPtr()))
    {
       return 1 ;    /*1  all indexes are disabled */
    }else return 0 ; /*0  indexes are not disabled */
}

THR_LOCK_DATA **mm_handler::store_lock(Session *,
                                    THR_LOCK_DATA **to,
                                    enum thr_lock_type lock_type)
{
  if (lock_type != TL_IGNORE && file->lock.type == TL_UNLOCK)
    file->lock.type=lock_type;
  *to++= &file->lock;
  return to;
}

/*
  We have to ignore ENOENT entries as the HEAP table is created on open and
  not when doing a CREATE on the table.
*/

int mm_handler::delete_table(const char * name)
{
   Session* session = ha_session();
   DatabaseManager* dbmgr = session->getDatabaseManager() ;
   int ret = 1;
   if( dbmgr){
     ret = dbmgr->dropTable(name);
   }
   return ret ;
}


int mm_handler::drop_table(const char * name)
{ /* drop the table */
  Session* session = ha_session();
  DatabaseManager* dbmgr = session->getDatabaseManager() ;
  int ret = 1;
  if( dbmgr){
    ret = dbmgr->dropTable(name);
  }
  close();
  return (ret) ;
}


int mm_handler::rename_table(const char * , const char * )
{
  //return heap_rename(from,to);
	return 0 ;
}


ha_rows mm_handler::records_in_range(uint32_t inx, key_range *,
                                  key_range *)
{
  KEY *key=table->key_info+inx;
  /*
  if (key->algorithm == HA_KEY_ALG_BTREE)
    return hp_rb_records_in_range(file, inx, min_key, max_key);

  if (!min_key || !max_key ||
      min_key->length != max_key->length ||
      min_key->length != key->key_length ||
      min_key->flag != HA_READ_KEY_EXACT ||
      max_key->flag != HA_READ_AFTER_KEY)
    return HA_POS_ERROR;     // Can only use exact keys
  */
  if (stats.records <= 1)
    return stats.records;

  /* Assert that info() did run. We need current statistics here. */
  //assert(key_stat_version == file->s->key_stat_version);
  return key->rec_per_key[key->key_parts-1];
}

int mm_handler::create(const char *name,
                       Table *table_arg,
                       HA_CREATE_INFO *create_info)
{
  int _error;
  if (table_arg->found_next_number_field)
  {
    //keydef[share->next_number_index].flag|= HA_AUTO_KEY;
    //found_real_auto_increment= share->next_number_key_offset == 0;
  }

  Session* session = ha_session();
  DatabaseManager* dbmgr = session->getDatabaseManager() ;
  if  (!dbmgr) _error   = ER_INVALID_SESSION_STATE ;
  else {
    DatabaseManagerImpl* dbmgrimpl = (DatabaseManagerImpl*) dbmgr ;
    /// create table.
    _error =  dbmgrimpl->createTable( name,*table_arg,create_info,internal_table) ;
  }

  return (_error);
}


void mm_handler::update_create_info(HA_CREATE_INFO *create_info)
{
   table->file->info(HA_STATUS_AUTO);
  if (!(create_info->used_fields & HA_CREATE_USED_AUTO))
    create_info->auto_increment_value= stats.auto_increment_value;
}

void mm_handler::get_auto_increment(uint64_t,
                                    uint64_t,
                                    uint64_t,
                                    uint64_t *first_value,
                                    uint64_t *nb_reserved_values)
{
  mm_handler::info(HA_STATUS_AUTO);
  *first_value= stats.auto_increment_value;
  /* such table has only table-level locking so reserves up to +inf */
  *nb_reserved_values= UINT64_MAX;
}


int mm_handler::cmp_ref(const unsigned char *ref1, const unsigned char *ref2)
{
  return memcmp(ref1, ref2, sizeof(unsigned char*));
}


bool mm_handler::check_if_incompatible_data(HA_CREATE_INFO *info_in,
                                            uint32_t table_changes)
{
  /* Check that auto_increment value was not changed */
  if ((info_in->used_fields & HA_CREATE_USED_AUTO &&
       info_in->auto_increment_value != 0) ||
      table_changes == IS_EQUAL_NO ||
      table_changes & IS_EQUAL_PACK_LENGTH) // Not implemented yet
    return COMPATIBLE_DATA_NO;
  return COMPATIBLE_DATA_YES;
}


